package com.pk.flink.scenario01.sink;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

public class NotFoundElementRedisSink extends RichSinkFunction<Tuple3<String, String, String>> {
    private Jedis jedis;

    @Override
    public void invoke(Tuple3<String, String, String> value, Context context) throws Exception {
        // 实现将 Tuple3 写入 Redis 的逻辑
        if(!jedis.isConnected())
        {
            jedis.connect();
        }

        jedis.hset(value.f0, value.f1, value.f2);
    }

    @Override
    public void close() throws Exception {
        super.close();

        if (jedis != null)
            jedis.close();
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        jedis = new Jedis("localhost", 6379, 10000);
        jedis.select(0);
    }
}